package com.wugh.fast.module.async.storage.impl;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.wugh.fast.module.async.dto.Task;
import com.wugh.fast.module.async.dto.TaskProcessing;
import com.wugh.fast.module.async.enums.TaskStatusEnum;
import com.wugh.fast.module.async.storage.TaskStorage;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * 说明: 基于redis实现的任务存储
 *
 * @author wugh
 * @date 2020/4/13 16:20
 */
public class RedisTaskStorage implements TaskStorage {

    public static final Duration TIME_OUT = Duration.ofDays(3);
    public static final String TASK_PREFIX = "fast.module.async.task:{}:detail";
    public static final String TASK_FINISHED_PREFIX = "fast.module.async.task:{}:finished";
    public static final String PROCESS_HASH_PREFIX = "fast.module.async.task:{}:process";

    private RedisTemplate<String, String> redisTemplate;

    public RedisTaskStorage(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * 查询任务信息和任务完成进度
     *
     * @param taskId
     * @return
     */
    @Override
    public Task queryTask(String taskId) {
        String data = redisTemplate.opsForValue().get(getTaskKey(taskId));
        if (StrUtil.isBlank(data)) {
            return null;
        }
        Task task = JSON.parseObject(data, Task.class);
        if (TaskStatusEnum.PROCESSING.getCode().equals(task.getStatus())) {
            String finished = redisTemplate.opsForValue().get(getTaskFinishedKey(taskId));
            task.setProcess(StrUtil.isBlank(finished) ? 0L : Long.parseLong(finished));
        }
        return task;
    }

    @Override
    public void insertTask(Task info) {
        redisTemplate.opsForValue().set(getTaskKey(info.getId()), JSON.toJSONString(info), TIME_OUT);
        redisTemplate.opsForValue().set(getTaskFinishedKey(info.getId()), Optional.ofNullable(info.getProcess()).orElse(0L) + "", TIME_OUT);
    }

    @Override
    public void updateTask(Task info) {
        redisTemplate.opsForValue().set(getTaskKey(info.getId()), JSON.toJSONString(info), TIME_OUT);
    }

    @Override
    public void deleteTask(String id) {
        redisTemplate.delete(getTaskKey(id));
    }

    @Override
    public void increaseProcess(String id) {
        redisTemplate.opsForValue().increment(getTaskFinishedKey(id));
    }

    @Override
    public void updateProcess(TaskProcessing process) {
        redisTemplate.opsForHash().put(getProcessKey(process.getTaskId()), process.getOperator(), JSON.toJSONString(process));
    }

    @Override
    public void deleteProcess(String taskId) {
        redisTemplate.delete(getTaskKey(taskId));
        redisTemplate.delete(getTaskFinishedKey(taskId));
        redisTemplate.delete(getProcessKey(taskId));
    }

    @Override
    public void deleteProcess(String taskId, String operator) {
        redisTemplate.opsForHash().delete(getProcessKey(taskId),operator);
    }

    @Override
    public List<TaskProcessing> queryProcessing(String taskId) {
        HashOperations<String, String, String> hash = redisTemplate.opsForHash();
        Map<String, String> entries = hash.entries(getProcessKey(taskId));
        return entries.values().stream()
                .filter(StrUtil::isNotBlank)
                .map(str -> JSON.parseObject(str, TaskProcessing.class))
                .collect(Collectors.toList());
    }

    private String getTaskKey(String taskId) {
        return StrUtil.format(TASK_PREFIX, taskId);
    }

    private String getTaskFinishedKey(String taskId) {
        return StrUtil.format(TASK_FINISHED_PREFIX, taskId);
    }

    private String getProcessKey(String taskId) {
        return StrUtil.format(PROCESS_HASH_PREFIX, taskId);
    }
}
